package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.cron.CronExpression;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.scheduler.MisfireStrategyEnum;
import com.xxl.job.admin.core.scheduler.ScheduleTypeEnum;
import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author 朱耀威
 */
public class JobScheduleHelper {

	private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);

	private static JobScheduleHelper instance = new JobScheduleHelper();

	public static JobScheduleHelper getInstance() {
		return instance;
	}

	public static final long PRE_READ_MS = 5000; // pre read

	private Thread scheduleThread;

	private Thread ringThread;

	private volatile boolean scheduleThreadToStop = false;

	private volatile boolean ringThreadToStop = false;

	private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

	public void start() {

		// schedule thread
		scheduleThread = new Thread(new Runnable() {
			@Override
			public void run() {

				try {
					TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
				}
				catch (InterruptedException e) {
					if (!scheduleThreadToStop) {
						logger.error(e.getMessage(), e);
					}
				}
				logger.info(">>>>>>>>> init xxl-job admin scheduler success.");

				// pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms,
				// qps = 1000/50 = 20)
				int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax()
						+ XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;

				while (!scheduleThreadToStop) {

					// Scan Job
					long start = System.currentTimeMillis();

					Connection conn = null;
					Boolean connAutoCommit = null;
					PreparedStatement preparedStatement = null;

					boolean preReadSuc = true;
					try {

						conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
						connAutoCommit = conn.getAutoCommit();
						conn.setAutoCommit(false);

						preparedStatement = conn.prepareStatement(
								"select * from xxl_job_lock where lock_name = 'schedule_lock' for update");
						preparedStatement.execute();

						// tx start

						// 1、pre read
						long nowTime = System.currentTimeMillis();
						List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao()
								.scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
						if (scheduleList != null && scheduleList.size() > 0) {
							// 2、push time-ring
							for (XxlJobInfo jobInfo : scheduleList) {

								// time-ring jump
								if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
									// 2.1、trigger-expire > 5s：pass && make
									// next-trigger-time
									logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());

									// 1、misfire match
									MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum
											.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
									if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
										// FIRE_ONCE_NOW 》 trigger
										JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null,
												null, null);
										logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = "
												+ jobInfo.getId());
									}

									// 2、fresh next
									refreshNextValidTime(jobInfo, new Date());

								}
								else if (nowTime > jobInfo.getTriggerNextTime()) {
									// 2.2、trigger-expire < 5s：direct-trigger && make
									// next-trigger-time

									// 1、trigger
									JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null,
											null);
									logger.debug(
											">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());

									// 2、fresh next
									refreshNextValidTime(jobInfo, new Date());

									// next-trigger-time in 5s, pre-read again
									if (jobInfo.getTriggerStatus() == 1
											&& nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {

										// 1、make ring second
										int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);

										// 2、push time ring
										pushTimeRing(ringSecond, jobInfo.getId());

										// 3、fresh next
										refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

									}

								}
								else {
									// 2.3、trigger-pre-read：time-ring trigger && make
									// next-trigger-time

									// 1、make ring second
									int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);

									// 2、push time ring
									pushTimeRing(ringSecond, jobInfo.getId());

									// 3、fresh next
									refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

								}

							}

							// 3、update trigger info
							for (XxlJobInfo jobInfo : scheduleList) {
								XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
							}

						}
						else {
							preReadSuc = false;
						}

						// tx stop

					}
					catch (Exception e) {
						if (!scheduleThreadToStop) {
							logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
						}
					}
					finally {

						// commit
						if (conn != null) {
							try {
								conn.commit();
							}
							catch (SQLException e) {
								if (!scheduleThreadToStop) {
									logger.error(e.getMessage(), e);
								}
							}
							try {
								conn.setAutoCommit(connAutoCommit);
							}
							catch (SQLException e) {
								if (!scheduleThreadToStop) {
									logger.error(e.getMessage(), e);
								}
							}
							try {
								conn.close();
							}
							catch (SQLException e) {
								if (!scheduleThreadToStop) {
									logger.error(e.getMessage(), e);
								}
							}
						}

						// close PreparedStatement
						if (null != preparedStatement) {
							try {
								preparedStatement.close();
							}
							catch (SQLException e) {
								if (!scheduleThreadToStop) {
									logger.error(e.getMessage(), e);
								}
							}
						}
					}
					long cost = System.currentTimeMillis() - start;

					// Wait seconds, align second
					if (cost < 1000) { // scan-overtime, not wait
						try {
							// pre-read period: success > scan each second; fail > skip
							// this period;
							TimeUnit.MILLISECONDS
									.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
						}
						catch (InterruptedException e) {
							if (!scheduleThreadToStop) {
								logger.error(e.getMessage(), e);
							}
						}
					}

				}

				logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
			}
		});
		scheduleThread.setDaemon(true);
		scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
		scheduleThread.start();

		// ring thread
		ringThread = new Thread(new Runnable() {
			@Override
			public void run() {

				while (!ringThreadToStop) {

					// align second
					try {
						TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
					}
					catch (InterruptedException e) {
						if (!ringThreadToStop) {
							logger.error(e.getMessage(), e);
						}
					}

					try {
						// second data
						List<Integer> ringItemData = new ArrayList<>();
						int nowSecond = Calendar.getInstance().get(Calendar.SECOND); // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
						for (int i = 0; i < 2; i++) {
							List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
							if (tmpData != null) {
								ringItemData.addAll(tmpData);
							}
						}

						// ring trigger
						logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = "
								+ Arrays.asList(ringItemData));
						if (ringItemData.size() > 0) {
							// do trigger
							for (int jobId : ringItemData) {
								// do trigger
								JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
							}
							// clear
							ringItemData.clear();
						}
					}
					catch (Exception e) {
						if (!ringThreadToStop) {
							logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
						}
					}
				}
				logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
			}
		});
		ringThread.setDaemon(true);
		ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
		ringThread.start();
	}

	private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
		Date nextValidTime = generateNextValidTime(jobInfo, fromTime);
		if (nextValidTime != null) {
			jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
			jobInfo.setTriggerNextTime(nextValidTime.getTime());
		}
		else {
			jobInfo.setTriggerStatus(0);
			jobInfo.setTriggerLastTime(0);
			jobInfo.setTriggerNextTime(0);
			logger.warn(
					">>>>>>>>>>> xxl-job, refreshNextValidTime fail for job: jobId={}, scheduleType={}, scheduleConf={}",
					jobInfo.getId(), jobInfo.getScheduleType(), jobInfo.getScheduleConf());
		}
	}

	private void pushTimeRing(int ringSecond, int jobId) {
		// push async ring
		List<Integer> ringItemData = ringData.get(ringSecond);
		if (ringItemData == null) {
			ringItemData = new ArrayList<Integer>();
			ringData.put(ringSecond, ringItemData);
		}
		ringItemData.add(jobId);

		logger.debug(
				">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData));
	}

	public void toStop() {

		// 1、stop schedule
		scheduleThreadToStop = true;
		try {
			TimeUnit.SECONDS.sleep(1); // wait
		}
		catch (InterruptedException e) {
			logger.error(e.getMessage(), e);
		}
		if (scheduleThread.getState() != Thread.State.TERMINATED) {
			// interrupt and wait
			scheduleThread.interrupt();
			try {
				scheduleThread.join();
			}
			catch (InterruptedException e) {
				logger.error(e.getMessage(), e);
			}
		}

		// if has ring data
		boolean hasRingData = false;
		if (!ringData.isEmpty()) {
			for (int second : ringData.keySet()) {
				List<Integer> tmpData = ringData.get(second);
				if (tmpData != null && tmpData.size() > 0) {
					hasRingData = true;
					break;
				}
			}
		}
		if (hasRingData) {
			try {
				TimeUnit.SECONDS.sleep(8);
			}
			catch (InterruptedException e) {
				logger.error(e.getMessage(), e);
			}
		}

		// stop ring (wait job-in-memory stop)
		ringThreadToStop = true;
		try {
			TimeUnit.SECONDS.sleep(1);
		}
		catch (InterruptedException e) {
			logger.error(e.getMessage(), e);
		}
		if (ringThread.getState() != Thread.State.TERMINATED) {
			// interrupt and wait
			ringThread.interrupt();
			try {
				ringThread.join();
			}
			catch (InterruptedException e) {
				logger.error(e.getMessage(), e);
			}
		}

		logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
	}

	// ---------------------- tools ----------------------
	public static Date generateNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
		ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(jobInfo.getScheduleType(), null);
		if (ScheduleTypeEnum.CRON == scheduleTypeEnum) {
			Date nextValidTime = new CronExpression(jobInfo.getScheduleConf()).getNextValidTimeAfter(fromTime);
			return nextValidTime;
		}
		else if (ScheduleTypeEnum.FIX_RATE == scheduleTypeEnum /*
																 * || ScheduleTypeEnum.
																 * FIX_DELAY ==
																 * scheduleTypeEnum
																 */) {
			return new Date(fromTime.getTime() + Integer.valueOf(jobInfo.getScheduleConf()) * 1000);
		}
		return null;
	}

}
